package cn.com.greatwall.kafka.service.impl;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.FutureTask;
import java.util.concurrent.TimeUnit;

import org.springframework.stereotype.Service;

import cn.com.greatwall.kafka.domain.Cluster;
import cn.com.greatwall.kafka.service.ClusterService;
import cn.com.greatwall.kafka.service.ZooKeeperService;
import cn.com.greatwall.kafka.service.dto.ClusterDto;
import cn.com.greatwall.kafka.util.ZKUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;

/**
 * @Author herw
 * @Time 2021-01-13 09:32:23
 * @Version 1.0
 * @Description: TODO(用一句话描述该文件做什么)
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class ZooKeeperServiceImpl implements ZooKeeperService {
    private final ClusterService clusterService;
    private Map<String, ZKUtil> cacheZookeeperMap = new HashMap<>(30);
    
    @Override
    public boolean testConnectZookeeper(Cluster resources) {
        boolean flag = false;
        FutureTask<Object> futureTask = new FutureTask<>(new Callable<Object>() {
            @Override
            public Object call() throws Exception {
                ZKUtil zooKeeper = null;
                try {
                    zooKeeper = new ZKUtil(resources.getZkAddress());
                } catch (Exception e) {
                    log.warn("connect zookper error.", e);
                }
                return zooKeeper;
            }
        });
        new Thread(futureTask).start();
        ZKUtil client = null;
        try {
            client = (ZKUtil) futureTask.get(3, TimeUnit.SECONDS);
            if (client != null) {
                flag = true;
            }
        } catch (Exception e) {
            log.warn("zookeeper is inactive.", e);
        } finally {
            try {
                if(client != null) {
                    client.close();
                }
            } catch (Exception e) {
                
            }
        }
        return flag;
    }

    @Override
    public ZKUtil getZooKeeper(String clusterId) {
        synchronized (cacheZookeeperMap) {
            ClusterDto cluster = clusterService.findById(Long.parseLong(clusterId));
            if(!cacheZookeeperMap.containsKey(clusterId)) {
                this.cacheZookeeperMap.put(clusterId, new ZKUtil(cluster.getZkAddress()));
            }
            return this.cacheZookeeperMap.get(clusterId);
        }
    }
}
